package com.niit.mobileDevide.age;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class AgeReducer extends Reducer<Text, AgeBean, NullWritable, Text> {

    @Override
    protected void reduce(Text key, Iterable<AgeBean> values, Context context) throws IOException, InterruptedException {
        Map<String, Integer> deviceModelCounts = new HashMap<>();

        // 统计每个设备型号的数量
        for (AgeBean val : values) {
            String deviceModel = val.getDeviceModel();
            deviceModelCounts.put(deviceModel, deviceModelCounts.getOrDefault(deviceModel, 0) + val.getCount());
        }

        // 按数量降序排序，并取前5个
        List<Map.Entry<String, Integer>> top5Devices = deviceModelCounts.entrySet().stream()
                .sorted((e1, e2) -> e2.getValue().compareTo(e1.getValue()))
                .limit(5)
                .collect(Collectors.toList());

        // 创建输出字符串
        StringBuilder outputBuilder = new StringBuilder();
        outputBuilder.append(key.toString()).append(": ");

        for (Map.Entry<String, Integer> entry : top5Devices) {
            outputBuilder.append(entry.getKey()).append("(").append(entry.getValue()).append("), ");
        }

        // 去掉最后一个逗号和空格
        if (outputBuilder.length() > 2) {
            outputBuilder.setLength(outputBuilder.length() - 2);
        }

        // 输出结果
        NullWritable nullKey = NullWritable.get(); // 获取 NullWritable 的实例
        context.write(nullKey, new Text(outputBuilder.toString()));
    }
}